package com.wyz.socket.core.impl.dispatcher;

import com.wyz.socket.core.base.IoArgs;
import com.wyz.socket.core.base.frame.Frame;
import com.wyz.socket.core.base.frame.ReceiveFrameFactory;
import com.wyz.socket.core.base.frame.receive.*;
import com.wyz.socket.core.base.packet.AbstractReceivePacket;

import java.io.Closeable;
import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import java.util.Collection;
import java.util.HashMap;


/**
 * 主要是 提供对ioArgs  的解析 ,分析 ,  消费
 * <p>
 * 1 提供 Ioargs 的基础是 能够解析 ioargs , 通过解析args 内的信息,提供大小合适的ioargs
 * 2 解析ioargs的基础是能够消费 args ,通过消费ioargs的信息,存储并对下一个ioargs 进行提供
 * 3 消费args 目的是为了提供下一个args 以及 合成frame 并最终合成packet
 * <p>
 * 初次提供args 判断当前是否有frame帧 , 若无 默认提供一个frame头大小的ioargs空间
 */
public class AsyncPacketWriter implements Closeable {

    private final PacketProvider provider;
    private final HashMap<Short, PacketModel> packetMap = new HashMap<>();
    private final IoArgs ioArgs = new IoArgs();
    private volatile Frame frame;

    public AsyncPacketWriter(PacketProvider provider) {
        this.provider = provider;
    }

    public synchronized void consumeIoArgs(IoArgs args) {
        if (frame == null) {
            AbstractReceiveFrame frameTemp;
            do {
                frameTemp = buildFrame(args); // 当前的frame为空 ,进行创建frame 主要对前6个字节进行分析
            } while (frameTemp == null && args.hasRemaining());

            if (frameTemp == null) { //  心跳
                return;
            }
            frame = frameTemp;
            if (!args.hasRemaining()) { // 取消
                return;
            }
        }
        Frame frameTemp = this.frame;
        do {
            try {
                if (frameTemp.handle(args)) {// 进行消费
                    if (frameTemp instanceof ReceiveHeaderFrame) {
                        ReceiveHeaderFrame headerFrame = (ReceiveHeaderFrame) frameTemp;
                        AbstractReceivePacket<?, ?> abstractReceivePacket = providerPacket(headerFrame);
                        addPacket(frameTemp.getBodyIdentifier(), abstractReceivePacket);
                    } else if (frameTemp instanceof ReceiveEntityFrame) {
                        ReceiveEntityFrame entityFrame = (ReceiveEntityFrame) frameTemp;
                        completeEntityFrame(entityFrame);
                    }
                    this.frame = null;
                    break;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        } while (args.hasRemaining());
    }

    public synchronized IoArgs providerIoArgs() {
        ioArgs.limit(frame == null ? Frame.FRAME_HEADER_LENGTH : frame.getUnConsumeLength());
        return ioArgs;
    }

    private AbstractReceiveFrame buildFrame(IoArgs args) {
        AbstractReceiveFrame instance = null;
        try {
            instance = ReceiveFrameFactory.createInstance(args); // 只获取IoArgs 头消息6个字节 4个字节代表长度 2字节代表类型 从而创建一个对应的接收帧
        } catch (Exception e) {
            return null;
        }
        if (instance instanceof CancelReceiveFrame) { // 如果是取消接收帧
            cancelReceivePacket(instance.getBodyIdentifier()); // 取消识别码对应的receivePacket 及其通道channel
        } else if (instance instanceof HeartbeatReceiveFrame) {
            heartbeatReceivePacket();
            return null;
        } else if (instance instanceof ReceiveEntityFrame) { // 如果是接收体帧 则获取到对应的channel 并绑定消费
            WritableByteChannel channel = getPacketChannel(instance.getBodyIdentifier());
            ((ReceiveEntityFrame) instance).bindPacketChannel(channel);
        }
        return instance;
    }

    private void heartbeatReceivePacket() {
        provider.onReceiveHeartbeat();
    }

    private AbstractReceivePacket<?, ?> providerPacket(ReceiveHeaderFrame headerFrame) {
        return provider.providerPacket(headerFrame.getPacketLength(), headerFrame.getPacketType(), headerFrame.getPacketInfo());
    }

    private synchronized void completeEntityFrame(ReceiveEntityFrame entityFrame) {
        short id = entityFrame.getBodyIdentifier();
        int bodyLength = entityFrame.getBodyLength();
        PacketModel packetModel = packetMap.get(id);
        if (packetModel != null) {
            packetModel.unReceiveLength -= bodyLength;
            if (packetModel.unReceiveLength <= 0) {
                provider.completeSendPacket(packetModel.packet, true);
            }
        }
    }

    private synchronized void addPacket(short identifier, AbstractReceivePacket<?, ?> packet) {
        packetMap.put(identifier, new PacketModel(packet));
    }

    private synchronized WritableByteChannel getPacketChannel(short bodyIdentifier) {
        PacketModel packetModel = packetMap.get(bodyIdentifier);
        if (packetModel == null) {
            return null;
        }
        return packetModel.channel;
    }

    private synchronized void cancelReceivePacket(short bodyIdentifier) {
        PacketModel packetModel = packetMap.get(bodyIdentifier);
        if (packetModel != null) {
            AbstractReceivePacket<?, ?> packet = packetModel.packet;
            provider.completeSendPacket(packet, false);
        }
    }


    @Override
    public synchronized void close() {
        Collection<PacketModel> values = packetMap.values();
        for (PacketModel m : values) {
            provider.completeSendPacket(m.packet, false);
        }
        packetMap.clear();
    }

    interface PacketProvider {

        AbstractReceivePacket<?, ?> providerPacket(int length, byte type, byte[] headerInfo);

        void completeSendPacket(AbstractReceivePacket<?, ?> packet, boolean isSucceed);

        void onReceiveHeartbeat();
    }

    static class PacketModel {
        final AbstractReceivePacket<?, ?> packet;
        final WritableByteChannel channel;
        volatile long unReceiveLength;

        public PacketModel(AbstractReceivePacket<?, ?> packet) {
            this.packet = packet;
            this.channel = Channels.newChannel(packet.open());
            this.unReceiveLength = packet.length();
        }
    }
}
